package com.moli.iov.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;

public abstract class KafkaBatchConsumerServer implements BatchAcknowledgingMessageListener<String,Object> {

    protected final Logger logger=LoggerFactory.getLogger("KafkaBatchConsumerServer");

    @Override
    public void onMessage(List<ConsumerRecord<String, Object>> consumerRecords, Acknowledgment acknowledgment) {
        for (ConsumerRecord<String,Object> record:consumerRecords){
            String topic = record.topic();
            String key = record.key();
            Object value =record.value();
            long offset = record.offset();
            int partition = record.partition();
        }
        manualHandle(consumerRecords,acknowledgment);
    }
    public abstract void manualHandle(List<ConsumerRecord<String, Object>> consumerRecords,Acknowledgment acknowledgment);

}
